package com.nageffer.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageffer.shortlink.project.common.constant.ShortLinkConstant;
import com.nageffer.shortlink.project.common.convention.exception.ServiceException;
import com.nageffer.shortlink.project.dao.entity.*;
import com.nageffer.shortlink.project.dao.mapper.*;
import com.nageffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import com.nageffer.shortlink.project.mq.producer.DelayShortLinkStatsProducer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.*;

import static com.nageffer.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;

/**
 * 短链接监控状态保存消息队列消费者
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final RedissonClient redissonClient;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;
    private final DelayShortLinkStatsProducer delayShortLinkStatsProducer;
    private final StringRedisTemplate stringRedisTemplate;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;

    /**
     * 监听消息
     */
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        //获取消息的id
        RecordId id = message.getId();

        //判断当前的这个消息流程是否正在执行
        if (!messageQueueIdempotentHandler.isMessageProcessed(id.toString())) {
            //解决幂等问题
            //判断当前的这个消息流程是否执行完成
            //这一步的意义：假如处理消息的时候服务器宕机了，那redis无法删除key，那就需要放入redis时设置过期时间。
            //这时mq会重试，如果只进行一次判断的话，就直接返回了，消息就丢失了。所以需要让他如果没完成就抛异常重试，如果完成了就返回。
            if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }

        try {
            //接收到生产者发送的消息
            Map<String, String> producerMap = message.getValue();
            String fullShortUrl = producerMap.get("fullShortUrl");
            if (StrUtil.isNotBlank(fullShortUrl)) {
                String gid = producerMap.get("gid");
                ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
                //处理消息
                actualSaveShortLinkStats(fullShortUrl, gid, statsRecord);
            }
            //stream里的消费的消息不会删除，手动删除，防止占内存
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (Throwable ex) {
            //假如消费失败了，redis存在标识了，此时mq又重试了但被返回了会出现问题，那就需要出现问题时删除redis
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            log.error("记录短链接监控消费异常", ex);
            throw new ServiceException("处理消息失败！");
        }
        //设置消息为完成状态
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }


    /**
     *  更新监控数据
     */
    public void actualSaveShortLinkStats(String fullShortUrl, String gid, ShortLinkStatsRecordDTO statsRecord) {
        fullShortUrl = Optional.ofNullable(fullShortUrl).orElse(statsRecord.getFullShortUrl());

        //加入读写锁
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();
        // rLock.lock();
        if (!rLock.tryLock()) {
            //没有得到锁，使用延迟队列在此更新监控信息
            delayShortLinkStatsProducer.send(statsRecord);
            return;
        }

        try {
            //如果gid传过来为空，就主动去查询数据库
            //这里的坑：要使用StrUtil.isBlank，因为消息队列传过来的gid有可能为""
            /*
            if (StrUtil.isBlank(gid)){
                LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                        .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
                ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
                gid = shortLinkGotoDO.getGid();
            }
            */

            //保证队列消费时读取最新Gid
            LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                    .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
            ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
            gid = shortLinkGotoDO.getGid();

            //统计小时和星期
            int hour = DateUtil.hour(new Date(), true);//获得指定日期的小时数部分
            Week week = DateUtil.dayOfWeekEnum(new Date());//获得指定日期是星期几
            int weekValue = week.getIso8601Value();

            //创建并插入短链接基础访问监控
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .hour(hour)
                    .weekday(weekValue)
                    .pv(1)
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)
                    .build();
            linkAccessStatsMapper.shortLinkStats(linkAccessStatsDO);

            //统计地区信息
            //使用高德的api：restapi.amap.com/v3/ip?key=您的key&ip=114.247.50.2
            Map<String, Object> localeMap = new HashMap<>();
            localeMap.put("key", statsLocaleAmapKey);
            localeMap.put("ip", statsRecord.getRemoteAddr());
            String localeResult = HttpUtil.get(ShortLinkConstant.AMAP_REMOTE_URL, localeMap);
            JSONObject jsonObject = com.alibaba.fastjson2.JSON.parseObject(localeResult);
            String infcode = jsonObject.getString("infocode");
            String actualProvince = "未知";
            String actualCity = "未知";
            if (StrUtil.isNotBlank(infcode) && StrUtil.equals(infcode, "10000")){ //10000代表正确响应码
                //province为[]，其他也为[]
                String province = jsonObject.getString("province");
                boolean provinceFlag = StrUtil.equals(province, "[]");
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .fullShortUrl(fullShortUrl)
                        .gid(gid)
                        .date(new Date())
                        .cnt(1)
                        .province(actualProvince = provinceFlag ? "未知":province)
                        .city(actualCity = provinceFlag ? "未知":jsonObject.getString("city"))
                        .adcode(provinceFlag ? "未知":jsonObject.getString("adcode"))
                        .country("中国")
                        .build();
                linkLocaleStatsMapper.shortLinkLocaleState(linkLocaleStatsDO);
            }

            //统计操作系统信息
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .cnt(1)
                    .os(statsRecord.getOs())
                    .build();
            linkOsStatsMapper.shortLinkOsState(linkOsStatsDO);

            //统计浏览器信息
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .cnt(1)
                    .browser(statsRecord.getBrowser())
                    .build();
            linkBrowserStatsMapper.shortLinkBrowserState(linkBrowserStatsDO);

            //统计设备信息
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .date(new Date())
                    .cnt(1)
                    .device(statsRecord.getDevice())
                    .build();
            linkDeviceStatsMapper.shortLinkDeviceState(linkDeviceStatsDO);

            //统计网络信息
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .cnt(1)
                    .network(statsRecord.getNetwork())
                    .build();
            linkNetworkStatsMapper.shortLinkNetworkState(linkNetworkStatsDO);

            //统计日志
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .fullShortUrl(fullShortUrl)
                    .gid(gid)
                    .user(statsRecord.getUv())
                    .ip(statsRecord.getRemoteAddr())
                    .browser(statsRecord.getBrowser())
                    .os(statsRecord.getOs())
                    .device(statsRecord.getDevice())
                    .network(statsRecord.getNetwork())
                    .locale(StrUtil.join("-", "中国", actualProvince, actualCity))
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);

            //短链接访问统计自增
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1, statsRecord.getUvFirstFlag() ? 1:0, statsRecord.getUipFirstFlag() ? 1:0);

            //统计当天短链接
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .gid(gid)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .todayPv(1)
                    .todayUv(statsRecord.getUvFirstFlag() ? 1:0)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1:0)
                    .build();
            linkStatsTodayMapper.shortLinkTodayState(linkStatsTodayDO);
        } finally { //这里把catch块删了，否则这里catch了，调用这个方法时就catch不到异常了，执行不了幂等性逻辑。
            rLock.unlock();
        }
    }

}
